package cn.doitedu.dao;

import cn.doitedu.cache.CacheManagerRedisImpl;
import cn.doitedu.pojo.CombineCondition;
import cn.doitedu.pojo.EventCondition;
import cn.doitedu.pojo.LogBean;
import cn.doitedu.utils.ConnectionUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @create: 2021-11-13 19:26
 * @author: 静如海的男人
 * @program: eagle-markting
 * 从clickhouse中查询历史行为数据
 **/

public class HistoryEventClickHouseDaoImpl implements HistoryEventDao {
    private Connection clickHouse;
    private CacheManagerRedisImpl jedis;

    //初始化方法   创建ClickHouse连接
    @Override
    public void init(ParameterTool parameterTool) throws Exception {

        clickHouse = ConnectionUtils.createClickHouse(parameterTool);

        jedis = new CacheManagerRedisImpl(parameterTool);
    }

    /**
     * @author: 静如海
     * @date: 2021/11/16
     * 从clickhouse中查询历史行为数据  并映射成数字或字母
     */
    @Override
    public String queryEventSequenceStr(LogBean bean, CombineCondition combineCondition, long startTime, long endTime) throws Exception {
        //在查询clickHouse之前,检查缓存是否可用

        //大key
        String bigKey = bean.getDeviceId() + ":" + combineCondition.getCacheId();
        //小key
        String smallKey = startTime + ":" + endTime;

        //①直接命中 缓存的数据 t4-t8  想要的数据t4-t8
        String data = jedis.getData(bigKey, smallKey);
        if (data != null) return data;

        //②部分命中 缓存的数据 t4-t8  想要的数据t4-t8|t9-t10
        Set<String> keys = jedis.getTimeRanges(bigKey);
        for (String key : keys) {
            //获取到的key 开始时间:结束时间
            String[] fields = key.split(":");
            long start = Long.parseLong(fields[0]);
            long end = Long.parseLong(fields[1]);
            //判断是否满足第二种情况
            if (startTime == start && endTime > end) {
                data = jedis.getData(bigKey, key);
                String str = queryEventSequenceStrByTimeRange(bean.getDeviceId(), combineCondition, end, endTime);
                //将数据缓存
                jedis.setValueEx(bigKey, smallKey, data + str, 3600 * 12);
                return data + str;

            }else if (startTime > start && endTime == end){
                //③部分命中  缓存的数据 t4-t8  想要的数据 t1-t4-t8

                data = jedis.getData(bigKey, key);
                String str = queryEventSequenceStrByTimeRange(bean.getDeviceId(), combineCondition, startTime, start);
                //将数据缓存
                jedis.setValueEx(bigKey, smallKey, str + data, 3600 * 12);
                return str + data;
            }
        }


        //如果不满足以上三种情况直接查clickhouse
        return queryEventSequenceStrByTimeRange(bean.getDeviceId(), combineCondition, startTime, endTime);
    }

    private String queryEventSequenceStrByTimeRange(String deviceId, CombineCondition combineCondition, long startTime, long endTime) throws SQLException {
        //事先定义好的sql
        String sql = combineCondition.getSql();
        //事先定义好的开始时间
        long setStartTime = startTime;
        //事先定义好的结束时间
        long setEndTime = endTime;

        //请求数据库  查询历史行为数据
        PreparedStatement preparedStatement = clickHouse.prepareStatement(sql);
        preparedStatement.setString(1, deviceId);
        preparedStatement.setLong(2, setStartTime);
        preparedStatement.setLong(3, setEndTime);
        //返回查询的  eventId
        ResultSet resultSet = preparedStatement.executeQuery();

        //事先定义好的历史规则
        List<String> targets = combineCondition.getEventConditions().stream().map(EventCondition::getEventId).collect(Collectors.toList());

        //将返回的历史数据映射成数字
        StringBuffer sb = new StringBuffer();
        while (resultSet.next()) {
            sb.append(targets.indexOf(resultSet.getString(1)) + 1);
        }

        //存入到redis
        jedis.setValueEx(deviceId+":"+combineCondition.getCacheId(), startTime+":"+endTime, sb.toString(), 3600 * 12);

        //返回结果
        return sb.toString();
    }

    @Override
    public void close() throws Exception {
        clickHouse.close();
    }
}
